package com.xiguanlezz.cn.consumer.handler;

import com.xiguanlezz.cn.InvokeMethodConst;
import com.xiguanlezz.cn.consumer.context.RpcContext;
import com.xiguanlezz.cn.consumer.future.RpcFuture;
import com.xiguanlezz.cn.protocol.RpcHeader;
import com.xiguanlezz.cn.protocol.RpcProtocol;
import com.xiguanlezz.cn.protocol.RpcRequest;
import com.xiguanlezz.cn.protocol.RpcResponse;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.net.SocketAddress;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;

/**
 * @Author：习惯了种种
 * @CreateDate：2024/11/4 16:26
 * @Description：
 */
@Slf4j
public class RpcConsumerHandler extends SimpleChannelInboundHandler<RpcProtocol<RpcResponse>> {
    @Getter
    private Channel remoteChannel;
    private SocketAddress remoteAddress;

    private Map<Long, RpcFuture> pendingResponseMap = new ConcurrentHashMap<>();

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
//        this.remoteChannel = ctx.channel();
        this.remoteAddress = ctx.channel().remoteAddress();
    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        super.channelRegistered(ctx);
        // 为什么保存和远端建立的channel需要在channelRegistered事件中，而不是channelActive中？
        // 因为客户端建立连接过程是：channelRegistered ==> connect ==> channelActive
        // 而这些步骤和主线程是并行的，就会导致connect结束后要往对端发送数据时，channelActive可能还未执行
        // 解决该问题有两个方法
        // 方法一：在channelRegistered中保存远端的channel
        // 方法二：可以在handler调用sendRequest方法前进行简单自旋
        this.remoteChannel = ctx.channel();
    }

    public RpcFuture sendRequest(RpcProtocol<RpcRequest> request, String invokeMethodType) throws ExecutionException, InterruptedException {
        if (!remoteChannel.isActive()) {
            return null;
        }

        switch (invokeMethodType) {
            case InvokeMethodConst.SYNC:
                return sendRequestSync(request);
            case InvokeMethodConst.ASYNC:
                return sendRequestAsync(request);
            case InvokeMethodConst.ONEWAY:
                return sendRequestOneway(request);
            default:
                return null;
        }
    }

    private RpcFuture sendRequestSync(RpcProtocol<RpcRequest> request) {
        RpcHeader rpcHeader = request.getHeader();
        long messageId = rpcHeader.getMessageId();
        // 创建future对象
        RpcFuture rpcFuture = RpcFuture.createRpcFuture(request);
        pendingResponseMap.put(messageId, rpcFuture);
        // channel#writeAndFlush(msg)会从TailContext一路往前传递
        remoteChannel.writeAndFlush(request);
        // AQS自旋获取pendingMap中的响应数据，获取到就返回。是由channelRead0中的线程唤醒的
        return rpcFuture;
    }

    private RpcFuture sendRequestAsync(RpcProtocol<RpcRequest> request) {
        RpcFuture rpcFuture = RpcFuture.createRpcFuture(request);
        RpcHeader rpcHeader = request.getHeader();
        long messageId = rpcHeader.getMessageId();
        // 同步和异步调用都是基于并发Map进行线程间通信的
        pendingResponseMap.put(messageId, rpcFuture);
        // 将future对象塞入上下文中
        RpcContext.getContext().setRpcFuture(rpcFuture);
        // channel#writeAndFlush(msg)会从TailContext一路往前传递
        remoteChannel.writeAndFlush(request);
        return null;
    }

    private RpcFuture sendRequestOneway(RpcProtocol<RpcRequest> request) {
        // channel#writeAndFlush(msg)会从TailContext一路往前传递
        remoteChannel.writeAndFlush(request);
        return null;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcProtocol<RpcResponse> response) throws Exception {
        if (Objects.isNull(response)) {
            return;
        }

        log.info("consumer received response: {}", response);
        String invokeMethodType = response.getBody().getInvokeMethodType();
        RpcHeader rpcHeader = response.getHeader();
        long messageId = rpcHeader.getMessageId();
        RpcFuture rpcFuture = pendingResponseMap.get(messageId);
        switch (invokeMethodType) {
            case InvokeMethodConst.SYNC:
            case InvokeMethodConst.ASYNC:
                if (Objects.nonNull(rpcFuture)) {
                    // 唤醒阻塞的线程
                    rpcFuture.done(response);
                }
                return;
            default:
                return;
        }
    }

    public void closeChannel() {
        if (Objects.nonNull(remoteChannel)) {
            remoteChannel.close();
        }
    }
}
